package demo.db.influxdb;

import com.alibaba.fastjson.JSON;
import com.influxdb.Cancellable;
import com.influxdb.annotations.Column;
import com.influxdb.annotations.Measurement;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.InfluxQLQueryApi;
import com.influxdb.client.QueryApi;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.Authorization;
import com.influxdb.client.domain.Bucket;
import com.influxdb.client.domain.BucketRetentionRules;
import com.influxdb.client.domain.InfluxQLQuery;
import com.influxdb.client.domain.Permission;
import com.influxdb.client.domain.PermissionResource;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import com.influxdb.query.InfluxQLQueryResult;
import demo.java.net.URLDemo;
import demo.java.time.format.DateTimeFormatterDemo;
import demo.vo.tsdb.DataPoint;
import demo.vo.tsdb.TsdbData;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BoundParameterQuery;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.influxdb.impl.InfluxDBResultMapper;
import org.json.JSONObject;
import org.junit.Test;
import org.springframework.format.annotation.NumberFormat;

import java.math.BigDecimal;
import java.time.Instant;
import java.time.Period;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

/**
 * https://github.com/influxdata/influxdb-client-java#writes-and-queries-in-influxdb-2x
 */
public class InfluxDbQueryDemo {

    private static char[] token = "my-token".toCharArray();
    private static String org = "my-org";
    private static String bucket = "my-bucket";

    private static String majorUrl = "http://192.168.3.66:8086";
    private static String minorUrl = "http://192.168.3.66:8086";
    private static String username = "southgate";
    private static String password = "wmxPpAkoST18";
    private static String database = "southgate";

    @Test
    public void testQuery1() {
        String command = "SELECT * FROM domain_metrics  order by time desc limit 2";
        command = "SELECT * FROM domain_nginx_metrics order by time desc limit 2";
        String dbName = "southgate";

        InfluxDB influxDB = InfluxDBFactory.connect(majorUrl, username, password);
        QueryResult results = influxDB.query(new Query(command, dbName));
        //results.getResults()是同时查询多条SQL语句的返回值，此处我们只有一条SQL，所以只取第一个结果集即可。
        QueryResult.Result oneResult = results.getResults().get(0);
        System.out.println(JSON.toJSONString(oneResult, true));
        List<QueryResult.Series> seriesList = oneResult.getSeries();
        if (seriesList == null) {
            return;
        }
        for (QueryResult.Series series : seriesList) {
            System.out.println(series);
        }
        influxDB.close();
    }


    /**
     * Inluxdb 数据转 TSDB 数据
     *
     * @param seriesList
     * @return
     */
    public static List<TsdbData> mapTsdbData(List<QueryResult.Series> seriesList) {
        if (seriesList == null || seriesList.isEmpty()) {
            return Collections.emptyList();
        }
        List<TsdbData> list = new ArrayList<>(seriesList.size());
        for (QueryResult.Series series : seriesList) {
            String name = series.getName();
            Map<String, String> tags = series.getTags();
            List<String> columns = series.getColumns();
            List<List<Object>> values = series.getValues();
            TsdbData data = new TsdbData();
            data.setMetric(name);
            data.setTags(tags);
            int idxTime = columns.indexOf("time");
            int idxQps = columns.indexOf("qps");
            List<DataPoint> dps = values.stream().map(v -> {
                Object time = v.get(idxTime);
                Object qps = v.get(idxQps);
                long ts = DateTimeFormatterDemo.parseUTC((String) time);
                return new DataPoint(ts, (Double) qps);
            }).collect(Collectors.toList());
            data.setDps(dps);
            list.add(data);
        }
        return list;
    }

    @Test
    public void testQuerySouthGate() {
        String influxQL0 = "SELECT * FROM domain_metrics order by time desc limit 10";
        String influxQL1 = "SELECT adomain,qps FROM domain_metrics order by time desc limit 10";
        String influxQL2 = "SELECT adomain,qps FROM domain_metrics where time >= '2023-03-12T00:00:00Z' and time <= '2023-03-13T00:00:00Z' limit 10";
        String influxQL3 = "SELECT adomain,qps FROM domain_metrics where time >= 1678380293275000000 and time <= 1678639493275000000 limit 10";
        String influxQL4 = "SELECT adomain,qps FROM domain_metrics where time >= $startTime and time <= $endTime limit 10";
        String influxQL5 = "SELECT adomain,qps FROM domain_metrics where time >= $startTime and time <= $endTime and adomain = $adomain limit 10";
        String influxQL6 = "SELECT sum(qps) as qps FROM domain_metrics where time >= $startTime and time <= $endTime and adomain = $adomain group by time(30s) fill(0) limit 10";
        String influxQL7 = "SELECT adomain,qps FROM domain_metrics where time >= $startTime and time <= $endTime group by adomain limit 10";
        String influxQL8 = "SELECT sum(qps)/30 as qps FROM domain_metrics where time >= $startTime and time <= $endTime group by adomain,time(30s) fill(0) limit 10";
        long endTime = TimeUnit.MILLISECONDS.toNanos(System.currentTimeMillis());
        long startTime = endTime - TimeUnit.MINUTES.toNanos(10);

        String influxQL = influxQL8;

        BoundParameterQuery query = BoundParameterQuery.QueryBuilder.newQuery(influxQL)
                .forDatabase(database)
                .bind("startTime", startTime)
                .bind("endTime", endTime)
                .bind("adomain", "adse.test.ximalaya.com")
                .create();

        System.out.println("Command = " + query.getCommand());
        System.out.println("ParameterJson = " + URLDemo.urlDecode(query.getParameterJsonWithUrlEncoded()));

        InfluxDB influxDB = InfluxDBFactory.connect(majorUrl, username, password);
        QueryResult results = influxDB.query(query);
        //results.getResults()是同时查询多条SQL语句的返回值，此处我们只有一条SQL，所以只取第一个结果集即可。
        QueryResult.Result oneResult = results.getResults().get(0);
        System.out.println("oneResult.json = " + JSON.toJSONString(oneResult, true));
        List<QueryResult.Series> seriesList = oneResult.getSeries();
        if (seriesList == null) {
            return;
        }
        for (QueryResult.Series series : seriesList) {
            System.out.println("Series = " + series);
        }
        InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
        List<DomainQps> metrics = resultMapper.toPOJO(results, DomainQps.class, "domain_metrics");
        System.out.println("metrics = " + JSON.toJSONString(metrics, true));

        List<TsdbData> tsdbDataList = mapTsdbData(seriesList);
        System.out.println("TsdbData = " + JSON.toJSONString(tsdbDataList, true));
        influxDB.close();
    }

    public static class DomainQps {

        @org.influxdb.annotation.Column(name = "time")
        String time;

        @org.influxdb.annotation.Column(name = "adomain")
        String domain;

        @NumberFormat
        @org.influxdb.annotation.Column(name = "qps")
        Double qps;

        public String getTime() {
            return time;
        }

        public void setTime(String time) {
            this.time = time;
        }

        public String getDomain() {
            return domain;
        }

        public void setDomain(String domain) {
            this.domain = domain;
        }

        public Double getQps() {
            return qps;
        }

        public void setQps(Double qps) {
            this.qps = qps;
        }
    }

    /**
     * 读数据
     */
    @Test
    public void testQuery() {
        InfluxDB influxDB = InfluxDBFactory.connect(majorUrl, username, password);
        String measureName = "api_metrics";
        String selectString = "";
        String retentionPolicy = "";
        String whereString = "";
        String groupString = "";
        long startTime = System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(10);
        long endTime = System.currentTimeMillis();
        StringBuilder sb = new StringBuilder(selectString)
                .append(" FROM ").append("\"").append(retentionPolicy).append("\".").append(measureName)
                .append(" WHERE ").append(whereString).append("time >= $startTime AND time <= $endTime")
                .append(groupString);
        Query query = BoundParameterQuery.QueryBuilder.newQuery(sb.toString()).forDatabase(database)
                .bind("startTime", startTime).bind("endTime", endTime).create();
        QueryResult queryResult = influxDB.query(query);
        InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
        List<JSONObject> alertServerKeyIndexList = resultMapper.toPOJO(queryResult, JSONObject.class, measureName);
        influxDB.close();
    }


    /**
     * For querying data we use QueryApi that allow perform synchronous, asynchronous and also use raw query response.
     * <p>
     * For POJO mapping, snake_case column names are mapped to camelCase field names if exact matches not found.
     * <p>
     * The synchronous query is not intended for large query results because the Flux response can be potentially unbound.
     */
    public void testSynchronousQuery() {
        InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", token, org, bucket);
        String flux = "from(bucket:\"my-bucket\") |> range(start: 0)";

        QueryApi queryApi = influxDBClient.getQueryApi();

        List<FluxTable> tables = queryApi.query(flux);
        for (FluxTable fluxTable : tables) {
            List<FluxRecord> records = fluxTable.getRecords();
            for (FluxRecord fluxRecord : records) {
                System.out.println(fluxRecord.getTime() + ": " + fluxRecord.getValueByKey("_value"));
            }
        }

        influxDBClient.close();
    }

    /**
     * The synchronous query offers a possibility map FluxRecords to POJO:
     */
    @Test
    public void testMapFluxRecords2POJO() {
        InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", token, org, bucket);
        String flux = "from(bucket:\"my-bucket\") |> range(start: 0) |> filter(fn: (r) => r._measurement == \"temperature\")";

        QueryApi queryApi = influxDBClient.getQueryApi();

        // Map to POJO
        List<Temperature> temperatures = queryApi.query(flux, Temperature.class);
        for (Temperature temperature : temperatures) {
            System.out.println(temperature.location + ": " + temperature.value + " at " + temperature.time);
        }

        influxDBClient.close();
    }

    /**
     * The Asynchronous query offers possibility to process unbound query and allow user to handle exceptions,
     * stop receiving more results and notify that all data arrived.
     */
    public void testAsynchronousQuery() throws InterruptedException {
        InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", token, org, bucket);
        // Query data
        String flux = "from(bucket:\"my-bucket\") |> range(start: 0)";

        QueryApi queryApi = influxDBClient.getQueryApi();

        queryApi.query(flux, new BiConsumer<Cancellable, FluxRecord>() {
            @Override
            public void accept(Cancellable cancellable, FluxRecord fluxRecord) {
                // The callback to consume a FluxRecord.
                // cancelable - object has the cancel method to stop asynchronous query
                System.out.println(fluxRecord.getTime() + ": " + fluxRecord.getValueByKey("_value"));
            }
        }, throwable -> {
            // The callback to consume any error notification.
            System.out.println("Error occurred: " + throwable.getMessage());
        }, () -> {
            // The callback to consume a notification about successfully end of stream.
            System.out.println("Query completed");
        });
        Thread.sleep(5_000);
        influxDBClient.close();
    }


    /**
     * The Raw query allows direct processing original CSV response:
     */
    public void testRawQuery() throws InterruptedException {
        InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", token, org);

        // Query data
        String flux = "from(bucket:\"my-bucket\") |> range(start: 0)";

        QueryApi queryApi = influxDBClient.getQueryApi();

        String csv = queryApi.queryRaw(flux);

        System.out.println("CSV response: " + csv);

        influxDBClient.close();
    }

    /**
     * The Asynchronous version allows processing line by line:
     */
    public void testRawQueryAsynchronous() throws InterruptedException {
        InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", token, org);

        //
        // Query data
        //
        String flux = "from(bucket:\"my-bucket\") |> range(start: 0)";

        QueryApi queryApi = influxDBClient.getQueryApi();

        queryApi.queryRaw(flux, (cancellable, line) -> {

            //
            // The callback to consume a line of CSV response
            //
            // cancelable - object has the cancel method to stop asynchronous query
            //
            System.out.println("Response: " + line);
        });

        Thread.sleep(5_000);

        influxDBClient.close();
    }

    /**
     * The Asynchronous version allows processing line by line:
     */
    public void testParameterizedQuery() throws InterruptedException {
        String url = "https://us-west-2-1.aws.cloud2.influxdata.com";
        String token = "my-token";
        String org = "my-org";
        String bucket = "my-bucket";
        try (InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org, bucket)) {

            QueryApi queryApi = client.getQueryApi();

            Instant yesterday = Instant.now().minus(Period.ofDays(1));

            Point p = Point.measurement("temperature")
                    .addTag("location", "north")
                    .addField("value", 60.0)
                    .time(yesterday, WritePrecision.NS);

            WriteApiBlocking writeApi = client.getWriteApiBlocking();
            writeApi.writePoint(p);

            //
            // Query range start parameter using Instant
            //
            Map<String, Object> params = new HashMap<>();
            params.put("bucketParam", bucket);
            params.put("startParam", yesterday.toString());

            String parametrizedQuery = "from(bucket: params.bucketParam) |> range(start: time(v: params.startParam))";

            List<FluxTable> query = queryApi.query(parametrizedQuery, org, params);
            query.forEach(fluxTable -> fluxTable.getRecords()
                    .forEach(r -> System.out.println(r.getTime() + ": " + r.getValueByKey("_value"))));

            //
            // Query range start parameter using duration
            //
            params.put("startParam", "-1d10s");
            parametrizedQuery = "from(bucket: params.bucketParam) |> range(start: duration(v: params.startParam))";
            query = queryApi.query(parametrizedQuery, org, params);
            query.forEach(fluxTable -> fluxTable.getRecords()
                    .forEach(r -> System.out.println(r.getTime() + ": " + r.getValueByKey("_value"))));

        }
    }

    public void testInfluxQL() {
        try (InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", token, org)) {

            // Query data
            String influxQL = "SELECT FIRST(\"free\") FROM \"influxql\"";

            InfluxQLQueryApi queryApi = influxDBClient.getInfluxQLQueryApi();

            // send request
            InfluxQLQueryResult result = queryApi.query(new InfluxQLQuery(influxQL, database)
                            .setPrecision(InfluxQLQuery.InfluxQLPrecision.SECONDS),
                    (columnName, rawValue, resultIndex, seriesName) -> {
                        // convert columns
                        switch (columnName) {
                            case "time":
                                return Instant.ofEpochSecond(Long.parseLong(rawValue));
                            case "first":
                                return new BigDecimal(rawValue);
                            default:
                                throw new IllegalArgumentException("unexpected column " + columnName);
                        }
                    });

            for (InfluxQLQueryResult.Result resultResult : result.getResults()) {
                for (InfluxQLQueryResult.Series series : resultResult.getSeries()) {
                    for (InfluxQLQueryResult.Series.Record record : series.getValues()) {
                        System.out.println(record.getValueByKey("time") + ": " + record.getValueByKey("first"));
                    }
                }
            }

        }
    }

    /**
     * InfluxDB 管理
     */
    @Test
    public void testManagement() {

        InfluxDBClient influxDBClient = InfluxDBClientFactory.create("http://localhost:8086", token);

        //
        // Create bucket "iot_bucket" with data retention set to 3,600 seconds
        //
        BucketRetentionRules retention = new BucketRetentionRules();
        retention.setEverySeconds(3600);

        Bucket bucket = influxDBClient.getBucketsApi().createBucket("iot-bucket", retention, "12bdc4164c2e8141");

        //
        // Create access token to "iot_bucket"
        //
        PermissionResource resource = new PermissionResource();
        resource.setId(bucket.getId());
        resource.setOrgID("12bdc4164c2e8141");
        resource.setType(PermissionResource.TYPE_BUCKETS);

        // Read permission
        Permission read = new Permission();
        read.setResource(resource);
        read.setAction(Permission.ActionEnum.READ);

        // Write permission
        Permission write = new Permission();
        write.setResource(resource);
        write.setAction(Permission.ActionEnum.WRITE);

        Authorization authorization = influxDBClient.getAuthorizationsApi()
                .createAuthorization("12bdc4164c2e8141", Arrays.asList(read, write));

        //
        // Created token that can be use for writes to "iot_bucket"
        //
        String token = authorization.getToken();
        System.out.println("Token: " + token);

        influxDBClient.close();
    }

    @Measurement(name = "temperature")
    public static class Temperature {

        @Column(tag = true)
        String location;

        @Column
        Double value;

        @Column(timestamp = true)
        Instant time;
    }
}
